package ct.analysis.io;

import ct.common.util.JDBCUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author Jeremy Zheng
 * @Date 2021/2/26 20:55
 * @Version 1.0
 * MySQL数据格式化输出对象
 */
public class MySQLRedisTextOutputFormat extends OutputFormat<Text,Text> {

    protected static class MySQLRecordWriter extends RecordWriter<Text, Text> {

        private Connection connection=null;
        Jedis jedis=null;

        public MySQLRecordWriter(){
            //获取资源
            connection= JDBCUtil.getConnection();
            jedis=new Jedis("hadoop101",6379);
        }

        //输出数据
        @Override
        public void write(Text key, Text value) throws IOException, InterruptedException {

            String[] values=value.toString().split("_");
            String sumCall=values[0];
            String sumDuration=values[1];
            PreparedStatement pstat=null;
            try {
                String insertSQL="insert into ct_call(telid,dateid,sumcall,duration) values(?,?,?,?)";
                pstat=connection.prepareStatement(insertSQL);
                String k = key.toString();
                String[] ks=k.split("_");

                String tel=ks[0];
                String date=ks[1];

                pstat.setInt(1,Integer.parseInt(jedis.hget("ct_user",tel)));
                pstat.setInt(2,Integer.parseInt(jedis.hget("ct_date",date)));
                pstat.setInt(3,Integer.parseInt(sumCall));
                pstat.setInt(4,Integer.parseInt(sumDuration));
                pstat.executeUpdate();
            } catch (SQLException e) {
                e.printStackTrace();
            }finally {
                if (pstat!=null){
                    try {
                        pstat.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }


            }
        }

        //释放资源
        @Override
        public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            if (connection!=null){
                try {
                    connection.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (jedis!=null){
                jedis.close();
            }

        }

    }

    @Override
    public RecordWriter<Text, Text> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        return new MySQLRecordWriter();
    }

    @Override
    public void checkOutputSpecs(JobContext jobContext) throws IOException, InterruptedException {

    }


    private FileOutputCommitter committer = null;

    public static Path getOutputPath(JobContext job) {
        String name = job.getConfiguration().get("mapreduce.output.fileoutputformat.outputdir");
        return name == null ? null : new Path(name);
    }

    @Override
    public OutputCommitter getOutputCommitter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        if (this.committer == null) {
            Path output = getOutputPath(taskAttemptContext);
            this.committer = new FileOutputCommitter(output, taskAttemptContext);
        }

        return this.committer;
    }
}
